package flume.source;

import org.apache.commons.io.FileUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 修改一个已经给在git上文件后,文件会变成蓝色 蓝色表示git文件已经被改过了，但是git云上没改 现在要同步
   从云上拉取下来 就是要拉
 *
 * Created by 王康 on 2018/7/10.
 * 自定义source
 * 是用来监听一个文件的Source
 * 实时监听一个文件 只要有新内容产生 就收集起来
 * 可以记录偏移量(读到哪一行) 如果flume出现问题 重启后 接着上一次的偏移量 继续读取
 */


public class TailFileSource extends AbstractSource implements Configurable, EventDrivenSource {

    private static final Logger logger = LoggerFactory.getLogger(TailFileSource.class);
    private String filePath;
    private String offsetFile;
    private String charset;
    private Long interval;
    private ExecutorService pool;
    private FileRunner fileRunner;


    /*
    * 最先调用configure 方法 且只调用一次
    *
    * */

    @Override
    public void configure(Context context) {
        //在其他地方也要使用这几个参数  就要 定义一个 全局的变量
         filePath = context.getString("filePath");
         offsetFile = context.getString("offsetFile");
        //如果设置 就取charset 如果没有设置 就以UTF-8为 默认值
         charset = context.getString("charset","UTF-8");
        //如果设置 就取interval 如果没有设置 就以1000L为 默认值
         interval = context.getLong("interval",1000L);
    }

    /*
    * 在configure方法调用之后 调用一次
    * */

    @Override
    public synchronized void start() {
        //固定的线程池 线程数量固定
        //Executors.newFixedThreadPool(10)
        //可缓冲的线程池 CPU切换速度很快 可以一直扩大
        //Executors.newCachedThreadPool()
        //因为不确定目录中有多少个File文件 所以线程池的个数也不好确定 创建一个线程池(只有一个线程 单线程)可以避免假死
        pool = Executors.newSingleThreadExecutor();
        //停掉的时候也要 停掉线程  所以 这个线程对象 不能够弄一个匿名内部类 最好弄一个静态内部类
        //创建一个实现Runnable 接口的实现类  还要拿到channel的 引用 方便下一步

        //调用父类的方法拿到Channelprocessor
        ChannelProcessor channelProcessor = getChannelProcessor();
        fileRunner = new FileRunner(filePath,offsetFile,charset,interval,channelProcessor);
        //将线程对象 放进去 提交线程池
        pool.submit(fileRunner);
        //执行父类中的方法
        super.start();


    }



    @Override
    public synchronized void stop() {
        //有时候不一定能停掉程序  首先停掉线程
        fileRunner.setFlag(false);
        //释放线程池
        pool.shutdown();
        super.stop();
    }

    //静态类不能读取 外部的非静态变量
    private static class FileRunner implements Runnable{
        //定义几个成员变量
        private String filePath;
        private String offsetFile;
        private String charset;
        private Long interval;
        private ChannelProcessor channelProcessor;
        private Long offset =0L;
        private RandomAccessFile raf;
        private File file;
        private boolean flag=true;



        //创建一个构造方法 将这些要用到的 成员变量 传入到构造方法中

        public FileRunner(String filePath, String offsetFile, String charset, Long interval,ChannelProcessor channelProcessor) {
            this.filePath = filePath;
            this.offsetFile = offsetFile;
            this.charset = charset;
            this.interval = interval;
            this.channelProcessor = channelProcessor;

            //---------------构造方法中的判断逻辑----------------//

            //判断是否存在偏移量文件 ？
            file = new File(offsetFile);
            if(!file.exists()){
                //常见偏移量文件 如果没有偏移量文件就创建一个文件
                try {
                    file.createNewFile();
                } catch (IOException e) {
                    //打印异常
                    logger.error("create offset file error",e);
                }
            }
            //如果有接着读
            try {
                //直接用FileUtils.readFileToString  读成String 类型
                String offsetStr = FileUtils.readFileToString(file);
                //判断一下 读取出来的字符串 不为空串 和 null
                if(offsetStr!=null && !"".equals(offsetStr)){
                    //将读取到的偏移量字符串转换成long 类型的值
                     offset = Long.parseLong(offsetStr);
                }
            } catch (IOException e) {
                logger.error("read offset file error",e);
            }
            //RandomAccessFile 的 seek 跳转到 指定的偏移量
            try {
                raf = new RandomAccessFile(filePath, "r");
                //指定偏移量
                raf.seek(offset);
            } catch (FileNotFoundException e) {
                logger.error("create log file error",e);
            } catch (IOException e) {
                logger.error("raf seek error",e);
            }
            //---------------构造方法中的判断逻辑----------------//
        }



        //run方法 在构造方法之后 在run之前  执行这个判断逻辑
        @Override
        public void run() {

            //不停的 监听 一个文件 可以用 while
            while(flag){
                //把读取的数据封装成Event
                try {
                    String line = raf.readLine();
                    //进行判断 是否为空 就是 读取文件的过程中没有新文件产生 会产生Null
                    if(line != null){
                        //RandomaccessFile seek过程中 会产生乱码问题 默认使用的编码集是ISO-8859-1
                        //如果想显示中文就进行转码 解决乱码问题 用 ISO-8859-1 将文件读出来 然后再用utf-8写进去
                        line  = new String(line.getBytes("ISO-8859-1"), charset);
                        //说明读到内容 封装成Event
                        Event event = EventBuilder.withBody(line.getBytes());
                        //封装好Event后  用ChannelProcessor 将数据发送到Channel中
                        channelProcessor.processEvent(event);
                        //获取最新的偏移量 然后写入到偏移量文件中
                         offset = raf.getFilePointer();
                        //将最新的偏移量写入文件中
                        FileUtils.writeStringToFile(file,offset + "");
                    }else{
                        //如果没有读到内容，就休息 间隔一段时间 再去探查这个文件是否有新的日志产生
                        Thread.sleep(interval);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        //设置flag方法
        public void setFlag(boolean flag) {
            this.flag = flag;
        }
    }

}
